package Flume;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collections;
import java.util.Properties;

public class Import {
    private static KafkaConsumer<String, String> consumer;
    private static Properties properties;
    public static void main(String[] args)   {
        //调用接收消息的方法
        try {
            receivemsg();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    static {
        properties = new Properties();
        properties.put("bootstrap.servers", "GTL:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id","aaa");
        properties.put("enable.auto.commit", true);

        consumer = new KafkaConsumer<String, String>(properties);
        consumer.subscribe(Collections.singleton("stuInfo"));
    }
    /**
     * 获取Kafka topic（test）上的数据
     */
    private  static void receivemsg() throws IOException {
//
        FileOutputStream fos = null;
        try {
            File file = new File("E:/hahaha.txt");
            fos = new FileOutputStream(file);
//             osw = new OutputStreamWriter(fos);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }

        while (true){
            //从Kafka读到了数据放在records
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String,String> r:records){
                System.out.printf("key=%s,value=%s,offet=%s,topic=%s",r.key() , r.value() , r.offset(), r.topic());
                String msg = r.topic()+r.offset()+r.key()+r.value()+"\r\n";

                fos.write(msg.getBytes());
                //jos.close();
                  fos.close();
              //  originalJar.close();
            }
        }


    }

}
